package com.kwin.demo.server.module.flink;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.kwin.demo.server.module.flink.configuration.FlinkProperty;
import com.kwin.demo.server.module.flink.util.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.springframework.util.CollectionUtils;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author kwin
 * @Date 2022/7/25 14:34
 **/
@Slf4j
public class FlinkStarter {

    private FlinkProperty flinkProperty;

    private SinkFunction sinkFunction;

    ExecutorService executorService = Executors.newFixedThreadPool(1);

    public FlinkStarter(FlinkProperty flinkProperty, SinkFunction sinkFunction) {
        this.flinkProperty = flinkProperty;
        this.sinkFunction = sinkFunction;
    }

    public void init() {
        executorService.submit(() -> {
            try {
                start();
            } catch (Exception e) {
                log.error("flink cdc start error", e);
            }
        });

    }
    //    @PostConstruct
    public void start() throws Exception {

        StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
        //设置job名称
        tEnv.getConfig().getConfiguration().setString("pipeline.name", flinkProperty.getPipelineName());

        if(CollectionUtils.isEmpty(flinkProperty.getMysqlDataSource())) {
            return;
        }
        for(FlinkProperty.MysqlSourceProperty mysqlSourceProperty : flinkProperty.getMysqlDataSource()) {
            SourceFunction<JSONObject> sourceFunction = generateMysqlSource(mysqlSourceProperty);
            DataStreamSource dataStreamSource = bsEnv.addSource(sourceFunction);
            dataStreamSource.addSink(sinkFunction);
        }
        //        SourceFunction<String> sourceFunction =
        //
        //        val env = StreamExecutionEnvironment.getExecutionEnvironment();
        //
        //        // 使用MySQLSource创建数据源
        //        // 同时指定StringDebeziumDeserializationSchema，将CDC转换为String类型输出
        //        val sourceFunction = MySQLSource.builder().hostname("your-ip").port(3306)
        //            .databaseList("demo").username("root").password("123456")
        //            .deserializer(new StringDebeziumDeserializationSchema).build();
        //
/*        DataStreamSource dataStreamSource = bsEnv.addSource(sourceFunction);
        //        bsEnv.addSource(sourceFunction2).print().setParallelism(1);
        dataStreamSource.addSink(new CustomSinkFunction());*/
        //
        bsEnv.execute();
    }

    private SourceFunction<JSONObject> generateMysqlSource(FlinkProperty.MysqlSourceProperty sourceProperty) {

        MySQLSource.Builder<JSONObject> mysqlSourceBuilder = MySQLSource.<JSONObject>builder();
        if(sourceProperty.getServerId() != null) {
            mysqlSourceBuilder.serverId(sourceProperty.getServerId());
        }

        mysqlSourceBuilder.hostname(sourceProperty.getHostname())
            .port(sourceProperty.getPort())
            .databaseList(sourceProperty.getDatabaseList())
            .username(sourceProperty.getUsername())
            .password(sourceProperty.getPassword())
            .tableList(sourceProperty.getTableList())
            .deserializer(new JsonDebeziumDeserializationSchema());
        return mysqlSourceBuilder.build();
    }


}
